#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include "StdThread.h"
#include "DoubleLinkList.h"
#include "LinkQueue.h"
#include "ThreadPool.h"
#define true 1
#define false 0
#define SleepTime 5

struct Task
{
    void *(*func)(void *);
    void *arg;
};
typedef struct Task task;

task *CreateTask(void *(*func)(void *), void *arg)
{
    task *t = (task *)malloc(sizeof(task));
    if (t == NULL)
    {
        printf("CreateTask malloc error\n");
        return NULL;
    }
    t->func = func;
    t->arg = arg;
    return t;
}
void FreeTask(task *t)
{
    free(t);
}

struct ThreadPool
{
    DLlist threads;    // 线程存放在链表中
    LQueue task_queue; // 任务存放在队列中，先来先处理

    int max_thread_num; // 最大线程数
    int min_thread_num; // 最小线程数

    int max_queue_size;  // 最大任务数量
    int busy_thread_num; // 正在工作的线程数量
    int exit_thread_num; // 需要退出的线程数量

    pthread_mutex_t pool_mutex;        // 线程池的锁
    pthread_mutex_t busy_thread_mutex; // 计数器的锁

    pthread_cond_t queue_not_empty; // 任务队列中有任务的条件
    pthread_cond_t queue_not_full;  // 任务队列不满的条件

    Thread *admin_thread; // 管理者线程
    int shutdown;         // 关闭标志位
};

void *thread_work(void *arg)
{
    ThreadP *p = (ThreadP *)arg;
    while (1)
    {
        pthread_mutex_lock(&p->pool_mutex); // 防止对一个线程抢夺一个任务，加锁
        // 如果任务队列为空并且线程池没有关闭,就让线程等待
        while (LQIsEmpty(&p->task_queue) == true && p->shutdown == false)
        {
            pthread_cond_wait(&p->queue_not_empty, &p->pool_mutex);
            if (p->exit_thread_num > 0)
            {
                p->exit_thread_num--;
                // 返回当前线程的线程号
                struct Node *TravelPoint = p->threads.head;
                while (TravelPoint != NULL)
                {
                    Thread *t = (Thread *)TravelPoint->data;
                    if (GetThreadId(t) == pthread_self())
                    {
                        RemoveByElement(&p->threads, t);
                        break;
                    }
                    TravelPoint = TravelPoint->next;
                }
                pthread_mutex_unlock(&p->pool_mutex);
                pthread_exit(NULL);
            }
        }
        // 如果关掉线程池,解锁，退线程
        if (p->shutdown == true)
        {
            pthread_mutex_unlock(&p->pool_mutex);
            pthread_exit(NULL);
        }
        // 从队列中取出任务
        task *tk = (task *)(*LQPop(&p->task_queue));
        // 取出任务后解锁
        pthread_mutex_unlock(&p->pool_mutex);
        // 拿走一个任务，就可以发送信号插入一个任务
        pthread_cond_broadcast(&p->queue_not_full);
        // 如果多个线程同时抢到了任务，那么忙的数量会发生错误，所以加锁
        pthread_mutex_lock(&p->busy_thread_mutex);
        p->busy_thread_num++;
        pthread_mutex_unlock(&p->busy_thread_mutex);

        // 执行任务
        tk->func(tk->arg);

        // 如果多个线程同时结束了任务，那么忙的数量会发生错误，所以加锁
        pthread_mutex_lock(&p->busy_thread_mutex);
        p->busy_thread_num--;
        pthread_mutex_unlock(&p->busy_thread_mutex);
        FreeTask(tk);
    }
    return NULL;
}

void *thread_manager(void *arg)
{
    ThreadP *p = (ThreadP *)arg;
    while (p->shutdown != true)
    {
        sleep(SleepTime);
        pthread_mutex_lock(&p->pool_mutex);
        // 任务多，线程不够
        int queueLen = GetQueueLen(&p->task_queue);
        int thread_num = GetDLlistLen(&p->threads);
        if (p->busy_thread_num < GetQueueLen(&p->task_queue) && thread_num < p->max_thread_num)
        {
            int add = (queueLen / 2) > (p->max_thread_num - thread_num) ? (p->max_thread_num - thread_num) : (queueLen / 2);
            for (int i = 0; i < add; i++)
            {
                Thread *t = InitThread(thread_work, p);
                InsertTail(&p->threads, t);
            }
        }
        // 线程多，任务少
        thread_num = GetDLlistLen(&p->threads);
        if (thread_num > p->busy_thread_num * 2 && thread_num > p->min_thread_num)
        {
            int minus_thread_num = (thread_num - p->busy_thread_num) / 2;
            int minus = (minus_thread_num) > (thread_num - p->min_thread_num) ? (thread_num - p->min_thread_num) : (minus_thread_num);
            p->exit_thread_num = minus;
            pthread_mutex_unlock(&p->pool_mutex);
            for (int i = 0; i < minus; i++)
            {
                pthread_cond_broadcast(&p->queue_not_empty);
            }
            continue;
        }
        pthread_mutex_unlock(&p->pool_mutex);
    }
    pthread_exit(NULL);
}

ThreadP *InitThreadPool(int max_thread_num, int min_thread_num, int max_queue_size)
{
    ThreadP *p = (ThreadP *)malloc(sizeof(ThreadP));
    if (p == NULL)
    {
        printf("InitThreadPool malloc error\n");
        return NULL;
    }
    InitDLlist(&p->threads);
    LQInit(&p->task_queue);

    p->min_thread_num = min_thread_num;
    p->max_thread_num = max_thread_num;
    p->max_queue_size = max_queue_size;
    p->busy_thread_num = 0;
    p->exit_thread_num = 0;

    pthread_mutex_init(&p->busy_thread_mutex, NULL);
    pthread_mutex_init(&p->pool_mutex, NULL);

    pthread_cond_init(&p->queue_not_full, NULL);
    pthread_cond_init(&p->queue_not_empty, NULL);

    p->shutdown = false;

    for (int i = 0; i < p->max_thread_num; i++)
    {
        Thread *t = InitThread(thread_work, p);
        InsertTail(&p->threads, t);
    }
    p->admin_thread = InitThread(thread_manager, p);

    return p;
}

void ThreadP_AddTask(ThreadP *p, void *(*func)(void *), void *arg)
{
    // 操作公共变量，需要上锁，如添加任务，线程正在取任务
    pthread_mutex_lock(&p->pool_mutex);
    // 如果任务队列满了，就让新的任务等待
    while (GetQueueLen(&p->task_queue) == p->max_queue_size)
    {
        pthread_cond_wait(&p->queue_not_full, &p->pool_mutex);
    }
    // 如果关闭线程池，解锁
    if (p->shutdown == true)
    {
        pthread_mutex_unlock(&p->pool_mutex);
        return;
    }
    // 将任务添加到任务队列中
    LQPush(&p->task_queue, CreateTask(func, arg));
    // 有任务了发送信号
    pthread_cond_broadcast(&p->queue_not_empty);
    pthread_mutex_unlock(&p->pool_mutex);
}

void DestoryThreadPool(ThreadP *p)
{
    if (p == NULL)
    {
        return;
    }
    p->shutdown = true;
    JoinThread(p->admin_thread);
    free(p->admin_thread);
    int len = GetDLlistLen(&p->threads);
    for (int i = 0; i < len; i++)
    {
        pthread_cond_broadcast(&p->queue_not_empty);
    }

    struct Node *TravelPoint = p->threads.head;
    while (TravelPoint != NULL)
    {
        Thread *t = (Thread *)TravelPoint->data;
        JoinThread(t);
        free(t);
        TravelPoint = TravelPoint->next;
    }

    FreeDLlist(&p->threads);

    while (LQIsEmpty(&p->task_queue) != true)
    {
        task *t = (task *)(*LQPop(&p->task_queue));
        free(t);
    }
    LQFree(&p->task_queue);

    pthread_mutex_destroy(&p->pool_mutex);
    pthread_mutex_destroy(&p->busy_thread_mutex);

    pthread_cond_destroy(&p->queue_not_empty);
    pthread_cond_destroy(&p->queue_not_full);
}